package com.runa.monitor.platinterface.mqtt;


import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
import com.runa.monitor.comm.entity.CommNetEquipment;
import com.runa.monitor.comm.enums.NetEquTypeEnum;
import com.runa.monitor.platinterface.base.BaseContainer;
import com.runa.monitor.platinterface.platService.NetEquService;
import com.runa.monitor.platinterface.platService.RepositoryService;
import lombok.extern.slf4j.Slf4j;
import lombok.val;
import okhttp3.Credentials;
import okhttp3.OkHttpClient;
import okhttp3.Request;
import okhttp3.Response;
import org.apache.commons.lang.exception.ExceptionUtils;
import org.springframework.scheduling.annotation.EnableScheduling;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Component;

import javax.annotation.Resource;
import javax.transaction.Transactional;
import java.util.ArrayList;
import java.util.HashSet;
import java.util.List;
import java.util.Objects;
import java.util.concurrent.TimeUnit;
import java.util.regex.Matcher;
import java.util.regex.Pattern;
import java.util.stream.Collectors;

@Component
@EnableScheduling
@Slf4j
public class EmqStatusSynProcessJob {

    @Resource
    private RepositoryService repositoryService;

    @Resource
    private NetEquService netEquService;


    @Scheduled(cron = "0 */2 * * * ?")
    @Transactional
    public void execute() {
        String regex = "\\d{1,3}\\.\\d{1,3}\\.\\d{1,3}\\.\\d{1,3}";
        Matcher matcher = Pattern.compile(regex).matcher(EmqConfig.getHOST());
        if (!matcher.find()) {
            log.error("网关mqtt相关配置解析错误------" + EmqConfig.getHOST());
            return;
        }
        OkHttpClient okHttpClient = new OkHttpClient.Builder()
                .connectTimeout(60, TimeUnit.SECONDS)
                .readTimeout(60, TimeUnit.SECONDS)
                .authenticator((route, response) -> {
                    String credential = Credentials.basic(EmqConfig.getHttpUserName(), EmqConfig.getHttpPassWord());
                    return response.request().newBuilder().header("Authorization", credential).build();
                })
                .build();
//        String requestUrl = "http://" + ip + ":8081/api/v4/clients?conn_state=connected";
        String requestUrl = EmqConfig.getHttpServer() + "/api/v4/clients?conn_state=connected";
//        log.debug("emq请求信息:" + requestUrl);
        log.info("emq请求信息:" + requestUrl);
        Request request = new Request.Builder().url(requestUrl).build();
        try {
            Response response = okHttpClient.newCall(request).execute();
            if (response.isSuccessful() && response.body() != null) {
                val responseString = Objects.requireNonNull(response.body()).string();
                log.debug("获取emq数据成功------" + responseString);
//                log.info("获取emq数据成功------" + responseString);
                JSONObject jsonObject = JSONObject.parseObject(responseString);
                JSONArray data = jsonObject.getJSONArray("data");
                //获取数据库数据直接比较
                List<CommNetEquipment> commNetEquipmentList = repositoryService.getCommNetEquipmentByType(NetEquTypeEnum.AI控制器);
                log.debug("获取commNetEquipmentList数据成功------" + JSON.toJSONString(commNetEquipmentList));
                HashSet<String> hashSet = commNetEquipmentList.stream().map(CommNetEquipment::getSerial).collect(Collectors.toCollection(HashSet::new));
                List<String> serial = new ArrayList<>();
                for (Object datum : data) {
                    JSONObject jsonObjectT = JSONObject.parseObject(datum.toString());
                    String clientid = jsonObjectT.get("clientid").toString();
                    if (clientid.startsWith("aigw_jy_")) clientid = clientid.replace("aigw_jy_", "");
                    if (clientid.startsWith("aigw_mt_")) clientid = clientid.replace("aigw_mt_", "");
                    if (clientid.startsWith("aigw_hmy_")) clientid = clientid.replace("aigw_hmy_", "");
                    if (clientid.startsWith("aigw_")) clientid = clientid.replace("aigw_", "");
                    log.debug("clientid: {}", clientid);
                    if (clientid.length() > 5 && Pattern.compile("[0-9]*").matcher(clientid).matches()) {
                        hashSet.remove(clientid);
                        netEquService.aiStormOnline(clientid);
                        serial.add(clientid);
                    }
                }
                BaseContainer.netStatusCallback().rccStatusChange(serial, true);
                BaseContainer.netStatusCallback().rccStatusChange(hashSet.stream().peek(s ->
                        netEquService.aiStormOffline(s)).collect(Collectors.toList()), false);
                response.body().close();
            } else {
                if (response.body() != null) {
                    response.body().close();
                }
            }
        } catch (Exception e) {
            log.error("获取emq数据处理失败: {}", ExceptionUtils.getFullStackTrace(e));
        }
    }


}
